package com.wz.jiangsu.controller.behavior;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.wz.jiangsu.bean.vo.show.ShowSecurityEventDataCommonWazuhDto;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @program: wz
 * @author: wangzhou
 * @create: 2024-01-31 10:39
 * @description:
 **/
@RestController
public class UserBehaviorController {


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    /**
     * 测试用户行为举止使用对象自带的 ts
     * @param dto
     * @param topic
     * @return
     */
    @PostMapping("/kafka/post/callbackTwo/getTs/userBehavior")
    public String sendCallbackOneMessagegetts(@RequestBody ShowSecurityEventDataCommonWazuhDto dto , @RequestParam("topic") String topic,@RequestParam("num") Integer num) {
        int count =0;

//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }

//            System.out.println("jsonStr = " + jsonStr);
//            System.out.println("message = " + jsonStr);
//        kafkaTemplate.send(topic, jsonStr).addCallback(new SuccessCallback<SendResult<String, Object>>() {
        for (int i = 0; i < num; i++) {

            long currentTimeMillis = System.currentTimeMillis();
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String timeStr = cn.hutool.core.date.DateUtil.format(new Date(currentTimeMillis),format);
            System.out.println("时间戳:" + currentTimeMillis + "," + "年月日:" + timeStr );
            dto.setTs(dto.getTs());
            dto.setUuid(IdUtil.randomUUID());
            String message = StrUtil.toString(dto);
            String jsonStr = JSONUtil.toJsonStr(dto);
            kafkaTemplate.send(topic, jsonStr).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();

//                    System.out.println("发送消息成功-主机日志:" + topic + "-" + partition + "-" + offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });
            count++;
            System.out.println("发射的总数：" + count);
        }


        return "bb";
    }
}
